package com.zhenhunfan;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * <p>Title: 一元RPC，用于演示FutureStub的使用</p>
 * <p>Description:</p>
 *
 * @author zhanghong
 * @date 2023/7/17
 */
public class TestGrpcClient1 {
    public static void main(String[] args) {
        //1.创建通信的管道
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000)
                .usePlaintext().build();
        try {
            //2.获得代理对象 stub
            TestServiceGrpc.TestServiceFutureStub testServiceFutureStub =
                    TestServiceGrpc.newFutureStub(managedChannel);
            //3.完成rpc调用
            //3.1 准备参数
            TestProto.TestRequest.Builder builder = TestProto.TestRequest.newBuilder();
            builder.setName("镇魂帆");
            TestProto.TestRequest helloRequest = builder.build();
            //3.2 进行功能rpc调用，获取响应的内容
            ListenableFuture<TestProto.TestResponse> testResponseFuture = testServiceFutureStub.test(helloRequest);

            //sync(testResponseFuture);

            async2(testResponseFuture);
            managedChannel.awaitTermination(1, TimeUnit.MINUTES);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            managedChannel.shutdown();
        }
    }

    /**
     * 同步操作
     *
     * @param testResponseFuture
     */
    public static void sync(ListenableFuture<TestProto.TestResponse> testResponseFuture) {
        try {
            TestProto.TestResponse testResponse = testResponseFuture.get();
            String result = testResponse.getResult();
            System.out.println("result:" + result);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 异步操作，简单演示，不能拿到响应数据
     *
     * @param testResponseFuture
     */
    public static void async(ListenableFuture<TestProto.TestResponse> testResponseFuture) {
        try {
            testResponseFuture.addListener(() -> {
                System.out.println("异步RPC响应回来了");
            }, Executors.newCachedThreadPool());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 异步操作2,可以拿到响应数据
     *
     * @param testResponseFuture
     */
    public static void async2(ListenableFuture<TestProto.TestResponse> testResponseFuture) {

        Futures.addCallback(testResponseFuture, new FutureCallback<TestProto.TestResponse>() {
            @Override
            public void onSuccess(TestProto.TestResponse testResponse) {
                System.out.println("响应数据为：" + testResponse.getResult());
            }

            @Override
            public void onFailure(Throwable throwable) {

            }
        }, Executors.newCachedThreadPool());
    }
}
